%%%-------------------------------------------------------------------
%%% File    : vect_ht.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created :  2 Jun 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(vect_ht).

-include("../include/util.hrl").
-include("../include/vect.hrl").
-include("../include/pipe.hrl").
-include("../include/iter.hrl").
-include("../include/vect_ht.hrl").


%% API
-export([ 
		 new/4,
		 new/5,
		 open/5,
		 close/1,
		 delete/2,
		 insert/2,
		 lookup/2,
		 delete_key/2,
		 all/1,
		 clear/1,
		 as_is/1,
		 count_logs/1]).
%% Tests
-export([ 
		 test/0,
		 test/3,
		 test1/0,
		 test2/0,
		 test3/0,
		 test4/0]).

-record(sess,
		{root_dir,
		 buckets_count,
		 ht_name,
		 view,
		 vect_arity,
		 lookup_type=log, %% possible values [log,index]
		 flush_threshold=20000
		}).

%%====================================================================
%% API
%%====================================================================
new(HtName,{Module,Func,Args}=_RootDirCall,Arity,BucketsCount) when
  is_atom(HtName),
  is_integer(BucketsCount),
  is_integer(Arity) ->
	new(HtName,apply(Module, Func, Args),Arity,BucketsCount);
%%--------------------------------------------------------------------
%% Function: 	new(RootDir,Arity,BucketsCount) -> vect()
%% Description:	Creates a descriptor of a vector implemented as disk 
%%				based hash table using 'disk_log'.
%%				RootDir - root directory where buckets reside. Last 
%%						  dir will be a name of a vector i.e. the tag 
%%						  of records stored.
%%				Arity - the number of elements in a tuple.
%%				BucketsCount - a number of buckets (logs) data 
%%				dispersed between.
%%--------------------------------------------------------------------
new(HtName,RootDir,Arity,BucketsCount) when 
  is_atom(HtName),
  is_list(RootDir),
  is_integer(BucketsCount),
  is_integer(Arity) ->
	new(HtName,RootDir,Arity,BucketsCount,#view{vect_name=HtName}).
%% new(HtName, {_Module,_Func,_Args}=RootDirCall,Arity,BucketsCount,View) when
%%   is_atom(HtName), 
%%   is_integer(BucketsCount),
%%   is_integer(Arity),
%%   is_record(View,view) ->
%% 	new(HtName,RootDirCall,Arity,BucketsCount,View);
new(HtName, RootDir,Arity,BucketsCount,View) when 
  is_atom(HtName) andalso 
  (is_list(RootDir) orelse (is_tuple(RootDir) andalso tuple_size(RootDir)==3)) andalso
  is_integer(BucketsCount) andalso
  is_integer(Arity) andalso
  is_record(View,view) ->
	#vect{
		 drv_module=?MODULE,
		 open_fun={open,[HtName,RootDir, Arity, BucketsCount,View]},
		 close_fun=close,
		 delete_fun=delete,
		 delete_key_fun=delete_key,
		 insert_fun=insert,
		 lookup_fun=lookup,
		 all_fun=all,
		 clear_fun=clear
		 }.
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
open(HtName,{M,F,A}, Arity, BucketsCount,View) when is_record(View,view) ->
	open(HtName,apply(M,F,A), Arity, BucketsCount,View);
open(HtName,RootDir, Arity, BucketsCount,View) when is_record(View,view) ->
	#sess{
		  root_dir=RootDir,
		  ht_name=HtName,
		  view=View,
		  vect_arity=Arity,
		  buckets_count=BucketsCount}.
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
close(Sess) when is_record(Sess,sess)->
	ok.
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
delete(Sess, Tuple) when is_record(Sess,sess),is_tuple(Tuple) ->
	delete(Sess, [Tuple]);
delete(Sess, Tuples) when is_record(Sess,sess),is_list(Tuples) ->
	Ets=ets:new('$tups_by_log', [bag,protected]),
	lists:foreach(fun(Tuple) ->
						  lists:foreach(fun(Elem) ->
												ets:insert(Ets, {key2file(Sess,Elem),Tuple})
										end,tuple_to_list(Tuple))
				  end,Tuples),
	delete_from_bucket(Sess,ets:first(Ets),Ets).
delete_from_bucket(_Sess,'$end_of_table',_Ets) ->
	ok;
delete_from_bucket(Sess,File,Ets) ->
	transform(Sess, File, 
			  fun(Tuples)->
					  lists:filter(fun({Tuple,_Timestamp})->
										   ets:match(Ets, {File,Tuple})==[]
								   end, Tuples)
			  end, "Delete tuples"),
	delete_from_bucket(Sess,ets:next(Ets, File),Ets).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
delete_key(Sess, Key) when is_record(Sess,sess) ->
	View=Sess#sess.view,
%% 	ExtFun=View#view.ext_fun,
	KeyPos=View#view.key_pos,
	Log=key2file(Sess,Key),
	TransFun=fun(Tuples) ->
					 lists:filter(
					   fun({Tuple0,_Timestamp})->
%% 							   Tuple=ExtFun(Tuple0),
							   element(KeyPos,Tuple0)/=Key
					   end, Tuples)
			 end,
	transform(Sess,Log,TransFun,"Delete by key").
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
insert(Sess, Tuples) when is_record(Sess,sess) andalso 
									 is_list(Tuples) ->
	insert(Sess, iter:from_list(Tuples));
insert(Sess, TupleIter) when is_record(Sess,sess) andalso 
								 is_record(TupleIter,iter) ->
	View=Sess#sess.view,
	VectName=View#view.vect_name,
	IntFun=View#view.int_fun,
	Arity=Sess#sess.vect_arity,
	ValidateIter=iter_mapper:new(
				   TupleIter, 
				   fun(Tuple) when is_tuple(Tuple),
								   tuple_size(Tuple)==Arity ->
						   IntFun(Tuple);
					  (Record) when is_tuple(Record),
								   	tuple_size(Record)==(Arity+1),
									element(1,Record)==VectName ->
						   [_|Tail]=tuple_to_list(Record),
						   IntFun(list_to_tuple(Tail));
					  (BadTuple) ->
						   Example=erlang:make_tuple(Arity,'_'),
						   Example1=erlang:make_tuple(Arity+1,'_'),
						   exit({vect_insert_bad_tuple,BadTuple,Sess,{expected_like,Example,'or',setelement(1, Example1, VectName)}})
				   end),
	Pipe=new_pipe_writer(Sess),
	Pump=pump:new(ValidateIter, Pipe),
	ok=pump:init(Pump);
insert(Sess, Tuple) when is_record(Sess,sess) andalso is_tuple(Tuple) ->
	insert(Sess, [Tuple]).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
lookup(Sess, Key) when is_record(Sess,sess) ->
	Log=key2file(Sess,Key),
	case filelib:is_regular(Log) of
		false ->
			iter:empty();
		true ->
			View=Sess#sess.view,
			KeyPos=View#view.key_pos,
			VectName=View#view.vect_name,
			ExtFun=View#view.ext_fun,
			LogIter=iter_disk_log:new([{name,{read_bucket,Log}},{file,Log},{mode,read_only}]),
			iter_mapper:new(LogIter, 
							fun({Tuple0,_Timestamp})->
									case element(KeyPos,Tuple0) of
										Key ->
											ext_tuple(VectName,Tuple0,ExtFun);
										_ ->
											?REJECT_REC
									end
							end)
	end.
ext_tuple(Tag,Tuple0,ExtFun) ->
	Tuple=ExtFun(Tuple0),
	list_to_tuple([Tag|tuple_to_list(Tuple)]).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
all(Sess) when is_record(Sess,sess) ->
	View=Sess#sess.view,
	VectName=View#view.vect_name,
	ExtFun=View#view.ext_fun,
	HtName=Sess#sess.ht_name,
	Arity=Sess#sess.vect_arity,
	BuckCount=Sess#sess.buckets_count,
	Root=Sess#sess.root_dir,
%% 	io:format("~p", [[Root,"/*/*/", HtName,".", Arity,".", BuckCount,".log"]]),
	ConcatIter=iter_concat:new([iter_disk_log:new([{name,{read_bucket,BucketFile}},{file,BucketFile},{mode,read_only}])
	 ||BucketFile<-filelib:wildcard([Root,"/*/*/", HtName,".", integer_to_list(Arity),".", integer_to_list(BuckCount),".log"])]),
	iter_mapper:new(ConcatIter,
							fun({Tuple0,_Timestamp})->
									ext_tuple(VectName, Tuple0, ExtFun)
							end).
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
clear(Sess) when is_record(Sess,sess) ->
	ok.
%%====================================================================
%% Internal functions
%%====================================================================
key2file(Sess,Key) ->
	HtName=Sess#sess.ht_name,
	RootDir=Sess#sess.root_dir,
	BuckCount=Sess#sess.buckets_count,
	Arity=Sess#sess.vect_arity,
	Ext=Sess#sess.lookup_type,
	KeyHashCode=erlang:phash2(Key,BuckCount),
	[A,B]=case term_to_binary(KeyHashCode) of
		<<_,97,B1>> ->
			[0,B1];
		<<_,98,0,0,B2,B1>> ->
			[B2,B1];
		_ ->
			exit({buckets_count_limit,65536,BuckCount})
	end,
    _File=lists:flatten(io_lib:format("~s/~2.16.0B/~2.16.0B/~p.~p.~p.~p",[RootDir,A,B,HtName,Arity,BuckCount,Ext])).


new_pipe_writer(Sess) ->
	#pipe{
			  param=Sess,
			  open_fun= fun pw_open/1,
			  push_fun= fun pw_push/2,
			  flush_fun= fun pw_flush/1,
			  close_fun= fun pw_close/1
		  }.

pw_open({Sess,terminal=_NextPipe}) ->
	{Sess,_Buff=dict:new(),Sess#sess.flush_threshold}.
pw_push(S,?EOF) ->
	S;
pw_push({_Sess,_Buff,0}=S,Item) when is_tuple(Item)->
	pw_push(pw_flush(S),Item);
pw_push({Sess,Buff,Count},Item) when is_tuple(Item)->
	View=Sess#sess.view,
	KeyPos=View#view.key_pos,
%% 	DestFiles=sets:to_list(sets:from_list([key2file(Sess,Key)||Key<-[element(KeyPos,Item)])),
	DestFiles=[key2file(Sess,Key)||Key<-[element(KeyPos,Item)]],
	Timestamp=now(),
	BuffOut=lists:foldl(fun(File,BuffIn)->
								dict:append(File, {Item,Timestamp}, BuffIn)
						end,
						Buff, DestFiles),
	{Sess,BuffOut,Count-1}.
pw_flush({Sess,Buff,_Count}) ->
	Bads=dict:fold(fun(File,Items,BadResults)->
					 case svc_dlmw:awrite_list(File, Items) of
						 {error,Reason} ->
							 [Reason|BadResults];
						 _ ->
							 BadResults
					 end
			  end,
			  [], Buff),
	case Bads of
		[] ->
			{Sess,_NewBuff=dict:new(),Sess#sess.flush_threshold};
		Err ->
			exit({flush_error,Err})
	end.
pw_close(_) ->
	ok.

transform(Sess,Log,TransformFun,Title) when is_function(TransformFun, 1)->
	try svc_dlmw:atransform(Log,TransformFun, Title, []) of
		JobPid ->
			MonRef=erlang:monitor(process, JobPid),
			receive
				{'DOWN', MonRef, process, JobPid, normal} ->
					ok;
				{'DOWN', MonRef, process, JobPid, noproc} ->
					ok;
				{'DOWN', MonRef, process, JobPid, ErrReason} ->
					throw(ErrReason)
			end
	catch
		_:{log_busy,Log,JobPid} ->
			?LOG_TRACE("Log busy by ~p",[JobPid]),
			ok=timer:sleep(100),
			transform(Sess,Log,TransformFun,Title)
	end.

as_is(X) ->
	X.

count_logs(Wildcard) ->
	FileNames=filelib:wildcard(Wildcard),
	lists:foldl(fun(Log,Count)->
						LogName={read,Log},
						{ok,LogName}=disk_log:open([{name,LogName},{file,Log},{mode,read_only}]),
						count_log(LogName,start,Count)
				end,0,FileNames).
count_log(Log,Cont,Count) ->
	case disk_log:chunk(Log,Cont) of
		{error, Reason} ->
			throw(Reason);
		{Cont2, Terms} ->
			count_log(Log,Cont2,Count+length(Terms));
		{Cont2, Terms, _Badbytes} ->
			count_log(Log,Cont2,Count+length(Terms));
		eof ->
			disk_log:close(Log),
			Count
	end.
%%-----------------------------------------------------------------------------
%% Test functions
%%-----------------------------------------------------------------------------

test() ->
	Vect=vect_ht:new("d:/_ht/spog",3, 10000),
	VSess=vect:open(Vect),
	ok=vect:insert(VSess,[{{"subject",X},{"predicate",Y},{"some big data, jewrgtfjqgfrfkubylq2eifqiwefiqwfgilqwfwqdfwqlebfbwqufqwfiwwiqlfiqwuiefbwqbwiqbfwkbferwgergwerhwerghe456347534y5ewgr43776435t4rwergwert534745fg",Z}}||X<-lists:seq(1, 100),Y<-lists:seq(1, 100),Z<-lists:seq(1, 100)]),
	vect:close(VSess).
test1() ->

	[spawn_link(fun()->
			  Vect=vect_ht:new("d:/_ht/spog",3, 10000),
			  VSess=vect:open(Vect),
			  iter:print(vect:lookup(VSess,{"subject",1})),
			  vect:close(VSess)
	  end)||X<-lists:seq(1, 4)].
test2() -> 
	ArgL=[{name,x},{file,"d:/_ht/spog/00/19/bucket.log"}],%,{mode,read_write}],
%% 	disk_log:open(ArgL),
	[spawn_link(fun()->
			  iter:print(iter_disk_log:new([{name,x},{file,"d:/_ht/spog/00/19/bucket.log"}]))%{opened,x}))
	  end)||X<-lists:seq(1, 4)].
test(SourceFile,VectRootDir,Buckets) -> 
	Vect=vect_ht:new(VectRootDir,3, Buckets),
	VSess=vect:open(Vect),
	ok=vect:insert(VSess,iter_term_reader:new_from_file(SourceFile)),
	vect:close(VSess).
test3() -> 
%% 	Vect=vect_ht:new(spog1,"d:/_ht/",3, 10000),
	Vect=svc_vms:get_vect('SPOG'),
	VSess=vect:open(Vect),
	ok=vect:insert(VSess,iter_term_reader:new_from_file("D:/_Development/Erlang/wsp/demetra/test/valg_bulk_SUITE_data/triple1.txt")),
	vect:close(VSess).
test4() -> 
	Vect=vect_ht:new("d:/_ht/spog2",3, 10000),
	VSess=vect:open(Vect),
	iter:print(vect:all(VSess)).